package com.tool_stdy.ConcurrencyModel.Disruptor;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import com.lmax.disruptor.*;

/**
 * Created by yan_li on 2016/11/11.
 */
public class DisruptorClient {
    static int BUFFER_SIZE = 1024 * 1024;
    static int THREAD_NUMBERS = 4;
    static ExecutorService executor = Executors.newFixedThreadPool(THREAD_NUMBERS);

    public static void main(String[] args) throws InterruptedException {
        EventFactory<Task> eventFactory = new EventFactory<Task>() {
            @Override
            public Task newInstance() {
                return new Task();
            }
        };
        /*
         * createSingleProducer创建一个单生产者的RingBuffer，
         * 第一个参数叫EventFactory，从名字上理解就是"事件工厂"，其实它的职责就是产生数据填充RingBuffer的区块。
         * 第二个参数是RingBuffer的大小，它必须是2的指数倍 目的是为了将求模运算转为&运算提高效率
         * 第三个参数是RingBuffer的生产都在没有可用区块的时候(可能是消费者（或者说是事件处理器） 太慢了)的等待策略
         */
        RingBuffer<Task> ringBuffer = RingBuffer.createSingleProducer(eventFactory, BUFFER_SIZE, new YieldingWaitStrategy());
        //创建SequenceBarrier
        SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();
        CountDownLatch latch = new CountDownLatch(5000);
        TradeHandler handler = new TradeHandler();
        handler.setCountDownLatch(latch);
        //创建消息处理器池
        WorkerPool<Task> workerPool = new WorkerPool<Task>(ringBuffer, sequenceBarrier, new IgnoreExceptionHandler(), handler);
        workerPool.start(executor);
        long start = System.currentTimeMillis();
        //下面这个生产8个数据
        for (int i = 0; i < 5000; i++) {
            long seq = ringBuffer.next();
            ringBuffer.get(seq).setPrice(i);
            ringBuffer.publish(seq);
        }
        latch.await();
        System.out.println("userd:" + (System.currentTimeMillis() - start));
        try {
            Thread.sleep(1000000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        workerPool.halt();//通知事件(或者说消息)处理器 可以结束了（并不是马上结束!!!）
        executor.shutdown();//终止线程
    }
}
